# -*- coding: utf-8 -*-
# @Author: ystlong
# @Date:   2018-11-25 13:18:49
# @Last Modified by:   ystlong
# @Last Modified time: 2018-11-26 23:02:03

import sys
import os
import http
import json
import time
import uuid

import proto

full_url = "http://172.17.0.2:5050/api/v1/scheduler"


class Scheduler(object):
    """docstring for Scheduler"""

    def __init__(self, task_num=0):
        super(Scheduler, self).__init__()
        self.framework_id = None
        self.streamid = None
        self.offers = None
        self.task_num = task_num
        self.accept_offer_id = None
        self.decline_num = task_num


    def schedule_task(self):
        if self.task_num > 0:
            self.accept()
            self.task_num -= 1
        # elif self.decline_num > 0:
        #     print "------------------"
        #     offers = self.offers["offers"]
        #     if offers[0].has_key("executor_ids"):
        #         print "***************"
        #         self.decline()
        #         self.decline_num -= 1

    def scheduler_heart(self, result, headers):
        """订阅framework后的持久连接事件处理回调

        处理master->scheduler的event

        Arguments:
            result {json_obj} -- [收到的响应事件信息]
            headers {dict} -- [响应头]

        Raises:
            Exception -- [收到ERROR事件时触发]
        """
        if result["type"] == "ERROR":
            print result["error"]["message"]
            raise Exception(result["error"]["message"])
        elif result["type"] == "SUBSCRIBED":
            self.framework_id = result["subscribed"]["framework_id"]["value"]
            self.streamid = headers["Mesos-Stream-Id"]
            print "------", result
        elif result["type"] == "OFFERS":
            self.offers = result["offers"]
            self.schedule_task()
            print "^^^^^^", result
        elif result["type"] == "HEARTBEAT":
            print "====", result
        elif result["type"] == "UPDATE":
            self.acknowledge(result["update"])
        else:
            print "******", result

    def teardown(self):
        """结束framework运行

        Raises:
            Exception -- [description]
        """
        if self.framework_id == None:
            raise Exception("no framework_id fetch")
        data = {
            "framework_id": {"value": self.framework_id},
            "type": "TEARDOWN"
        }
        datas = json.dumps(data)
        htp = http.HttpRequest(full_url)
        r = htp.post(data=datas, headers={"Mesos-Stream-Id": self.streamid})
        if r == 202:
            print "teardown : %s success" % (self.framework_id)

    def decline(self):
        decline_type = proto.DeclineType(self.framework_id)

        decline_type.add_offer_id(self.accept_offer_id)
        decline_type.add_offer_id(self.offers["offers"][0]["id"]["value"])

        d_struct = decline_type.json_serial()
        json_str = json.dumps(d_struct, indent=1)
        # json_str = json.dumps(a_struct)
        # print json_str
        request = http.HttpRequest(full_url)
        r = request.post(json_str, headers={"Mesos-Stream-Id": self.streamid})
        print "*********", r

    def acknowledge(self, update_info):
        status = update_info["status"]
        print "ack", status["executor_id"]["value"],
        print status["task_id"]["value"],
        print status["timestamp"],
        print status["container_status"]["network_infos"][0]["ip_addresses"][0]["ip_address"],
        print status["source"],
        print status["state"],
        print status["agent_id"]

        if status.has_key("uuid"):
            ack_type = proto.AcknowledgeType(
                self.framework_id,
                status["agent_id"]["value"],
                status["task_id"]["value"],
                status["uuid"]
                )
            json_str = json.dumps(ack_type.json_serial(), indent=1)
            request = http.HttpRequest(full_url)
            r = request.post(json_str, headers={"Mesos-Stream-Id": self.streamid})
            print "*********", r

    def accept(self):
        """接受master返回的资源，开始提交任务

        创建一个任务并提交
        """
        uuid_str = str(uuid.uuid4())
        print "==============", uuid_str

        sch_type = proto.SchedulerType("ACCEPT", self.framework_id)
        accept_type = proto.AcceptType()
        sch_type.type_value = accept_type

        self.accept_offer_id = self.offers["offers"][0]["id"]["value"]
        accept_type.offer_ids.append(proto.ValueType(self.accept_offer_id))

        opera_type = proto.LanuchOperation()
        accept_type.operations.append(opera_type)

        # task_info = proto.TaskInfo("test1", "task-id" + uuid_str)
        task_info = proto.TaskInfo("test1", uuid_str)
        opera_type.task_infos.append(task_info)

        task_info.agent_id = proto.ValueType(
            self.offers["offers"][0]["agent_id"]["value"])
        cmd_executor = proto.CommandExecutor(sys.argv[1])
        # cmd_executor.executor_id = proto.ValueType("eeeee")
        cmd_executor.executor_id = proto.ValueType(uuid_str)

        # cmd_executor.value = "sleep 1000"
        task_info.executor = cmd_executor

        cpu_recource = proto.Resource()
        cpu_recource.name = "cpus"
        cpu_recource.role = "*"
        cpu_recource.allocation_info = "*"
        cpu_recource.type = "SCALAR"
        cpu_recource.type_value = proto.ValueType(1.0)
        task_info.resources.append(cpu_recource)

        mem_recource = proto.Resource()
        mem_recource.name = "mem"
        mem_recource.role = "*"
        mem_recource.allocation_info = "*"
        mem_recource.type = "SCALAR"
        mem_recource.type_value = proto.ValueType(128.0)
        task_info.resources.append(mem_recource)

        a_struct = sch_type.json_serial()

        json_str = json.dumps(a_struct, indent=1)
        # json_str = json.dumps(a_struct)
        # print json_str

        request = http.HttpRequest(full_url)
        r = request.post(json_str, headers={"Mesos-Stream-Id": self.streamid})

    def subscribe(self, async=False):
        """订阅master信息
        mesos 通信第一步，通过其获得 mesos-stream-id
        请求成功时返回 200 code， 且响应头包含Transfer-Encoding: chunked

        Raises:
            Exception -- 当响应头不包含Transfer-Encoding: chunked触发
        """
        subscibe_type = proto.SubscribeType()
        framework_info = proto.FramworkInfoType()
        subscibe_type.framework_info = framework_info

        framework_info.name = "simple test framework"
        framework_info.user = "root"

        request = http.HttpRequest(full_url)
        headers = {"Connection": "keep-alive"}
        ret_code = request.post(subscibe_type.json_serial(), headers)

        rencoding = request.get_header("Transfer-Encoding")
        if rencoding.strip() == "chunked":
            if async == True:
                request.async_record_json(self.scheduler_heart, wait=True)
            else:
                request.sync_record_json(self.scheduler_heart)
        else:
            raise Exception(
                "not Transfer-Encoding: error msg [%s]" % (request.raw()))


sc = Scheduler(task_num=1)
sc.subscribe()

# hr = http.HttpRequest(full_url)
# # hr = http.HttpRequest("http://172.17.0.2:5050/state")
# r = hr.post(data=data, headers={"Connection": "keep-alive"})
# hr.async_record_json(sc.scheduler_heart)

# while True:
#     if sc.framework_id == None and sc.offers == None:
#         time.sleep(0.1)
#         continue
#         time.sleep(100)
#     print sc.framework_id
#     time.sleep(1)
#     # try:
#     #     sc.accept()
#     # except Exception as e:
#     #     print repr(e)
#     #     pass
#     sc.teardown()
#     break
# print r
# r = hr.raw()
# print r
# r = hr.json()
# print r
# r = hr.recordio()
# print r
# r = hr.record_json()
# print r
